package com.study.flink.example.hotitem

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.io._

/**
  * 生成用户行为数据
  *
  * @author stephen
  * @date 2019-07-23 14:52
  */
object UserBehaviorProducer {

  def main(args: Array[String]): Unit = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("auto.offset.reset", "latest")

    val fileUrl = HotItems.getClass.getClassLoader.getResource("UserBehavior.csv")
    val source = Source.fromFile(fileUrl.toURI)
    val producer = new KafkaProducer[String, String](properties)

    for (line <- source.getLines()) {
      producer.send(new ProducerRecord[String, String]("behavior", line))
      println(line)
      Thread.sleep(200)
    }

    producer.close()
  }
}
